/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */

package TorrentEngine;
import Logger.MTLogger;
import java.util.Vector;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.OutputStream;

/**
 *
 * @author Dhaval Moliya
 */
public class ThreadedPeerSender implements PeerSenderInterface, Runnable  {

    public final static int DataQueue = 0;
    public final static int ControlQueue = 1;

    OutputStream outputStream;
    Queue dataQueue;
    Queue controlQueue;
    Thread t;
    Object lock;
   
    ThreadedPeerSender()
    {
        lock=new Object();
    	dataQueue=new Queue(DataQueue);
    	controlQueue=new Queue(ControlQueue);
        //dataQueue.start();
       t = new Thread(this);
       t.start();
        //controlQueue.start();
    }
    public int queueControlMessage(byte[] byteArray) throws Exception {
        //throw new UnsupportedOperationException("Not supported yet.");
    	//System.out.println("Q Cntrl msg");
    	if(outputStream == null)
            return 1;
        try {
              //System.out.println("Inside QueueControlMessage \n");
        	controlQueue.enqueue(byteArray);
                //controlQueue.dequeue();
                //byteArray=(byte[]) controlQueue.dequeue();
        	//outputStream.write(byteArray);
                //outputStream.flush();
          } catch (IOException ex) {
               ex.printStackTrace();
               
          }
        lock.notifyAll();
        return 0;
    }

   public int queueDataMessage(int aPieceIndex, int aBegin, int aLength, MTTorrent torrent, MTPeer peer) throws IOException, Exception {
        //throw new UnsupportedOperationException("Not supported yet.");
    	MTPiece piece = torrent.piece(aPieceIndex);

        if(outputStream == null)
            return 1;

        if(piece==null) return 1;

        byte[] block = piece.getBlock(aBegin, aLength);

        log("Processing piece request " + aPieceIndex + " Begin: " + aBegin + " Length: " + aLength + " while piece totalsize: " + piece.getTotalSize(), peer);

        if(aBegin + aLength > piece.getTotalSize())
        {
            //close("Bad PIECE request (index is out of bounds)");
            return 2;
        }

        if(block == null)
        {
            //close("Failed to extract block of piece");
            return 3;
        }

        dataQueue.enqueue(new DataMessageTemplate(aPieceIndex, aBegin, aLength, torrent, peer));
        lock.notifyAll();
        return -1;
    }

    public void setOutputStream(OutputStream opStream) throws Exception{
       // throw new UnsupportedOperationException("Not supported yet.");

        outputStream=opStream;
        controlQueue.setOPS(opStream);

    }

     public void log(String aText, MTPeer peer)
        {
            MTLogger.writeLine("[Peer " + peer.getAddress() + ":" + peer.getPort() +
                " " + peer.getClient() + "]");
            MTLogger.writeLine("\t" + aText);
        }
    public void run() {
      //System.out.println("New Thread!");
     while(true) {
        //System.out.println("Inside While Loop");
	if(!controlQueue.queue.isEmpty())
		try {
                controlQueue.dequeue();
            } catch (Exception ex) {
                ex.printStackTrace();
            }
        //else
            //System.out.println("___________Empty____Queue________");

     if(!dataQueue.queue.isEmpty())
		try {
                dataQueue.dequeue();
            } catch (Exception ex) {
                ex.printStackTrace();
            }
            try {
                lock.wait();
            } catch (InterruptedException ex) {
                ex.printStackTrace();
            }
	}
    }

}
